package cn.doitedu.dw.etl

import org.apache.commons.lang3.StringUtils
import org.apache.log4j.{Level, Logger}
import org.apache.spark.sql.SparkSession

/**
 * 每日滚动更新： 设备和账号的绑定评分（表）
 *
 * 设备id,登录账号,绑定得分,最近一次登录时间戳
 * deviceid,account,score,ts
 *
 * create table dws18.device_account_bind(
 * deviceid  string,
 * account   string,
 * score     double,
 * ts        bigint
 * )
 * partitioned by ( dt string )
 * stored as parquet
 * ;
 *
 *
 *
 * 这个数据的加工，应该是一个逐日滚动计算的方式
 *
 * 具体计算逻辑：
 *     1. 加载 T-1 日的绑定评分表
 *     2. 从T日的ods增量数据中心，计算 T日的  "设备-账号绑定评分"
 *     3. 综合 T-1日 和 T日的  绑定评分数据得到 T日的绑定评分数据最终结果
 *
 *
 * T日的设备-账号绑定评分，得分规则：  每登录一次，+100分
 *
 * 两日数据综合合并的逻辑
 * T-1 日
 * d01,a01,900
 * d02,a02,800
 * d01,a11,600
 * d03,a03,700
 * d04,a04,600
 *
 * T日
 * d01,a01,200
 * d03,a03,100
 * d03,a13,100
 * d06,a06,200
 * d06,a04,100
 *
 * T日  FULL JOIN T-1日 的绑定评分表，条件是：  deviceid=deviceid,account=account
 *
 * 另行处理当日的没有任何账号关联的设备id
 * d05,\n ,200
 * d04,\n ,200
 *
 */
object DeviceIdAccountBind {

  def main(args: Array[String]): Unit = {

    // Logger.getLogger("org.apache").setLevel(Level.ERROR)

    if(args.size<3) {
      println(
        """
          |usage:
          |  args(0): 运行模式
          |  args(1): 评分表的T-1日期
          |  args(2): 日志表的当天日期
          |
          |""".stripMargin)
      sys.exit(1)
    }

    val spark = SparkSession.builder()
      .appName(DeviceIdAccountBind.getClass.getSimpleName)
      .enableHiveSupport()
      .master(args(0))
      .getOrCreate()


    // 真实数据： 加载T-1日的绑定评分表
    val preDayBind = spark.read
      .table("dws18.device_account_bind")
      .where(s"dt='${args(1)}'")

    // 缓存RDD，提高重复利用该RDD的效率
    preDayBind.cache()

    // 测试数据： 加载T-1日的绑定评分表
    /*val preDayBind = spark.read
      .option("header", "true")
      .csv("file:///D:/datayi18/data/idbind/day01_bind/day01_bind.txt")
      .selectExpr("deviceid", "account", "cast(timestamp as bigint) as timestamp", "cast(score as double) as score")
    preDayBind.show(100, false)*/



    val isBlank = (s:String)=>{
      StringUtils.isBlank(s)
    }
    spark.udf.register("isblank",isBlank)

    // 真实数据： 计算T日数据中的绑定得分
    val curDayLog = spark.read
      .table("ods18.event_app_log")
      .where(s"dt='${args(2)}'")
      .selectExpr("deviceid", "if(isblank(account),null,account) as account", "timestamp", "sessionid")

    // 测试数据： 计算T日数据中的绑定得分
    /*val curDayLog = spark.read
      .option("header", "true")
      .csv("file:///D:/datayi18/data/idbind/day02_log/day02.log")
      .selectExpr("deviceid", "account", "cast(timestamp as bigint) as timestamp", "sessionid")
    */
    curDayLog.createTempView("cur_day_log")

    /** **
     * d01,a01,12235,s01
     * d01,a01,12266,s01
     * d01,a01,12345,s02
     * d01,a01,12376,s02
     * d01,   ,12345,s03
     * d01,   ,12376,s03
     * d02,a02,12345,s04
     * d02,a02,12376,s04
     * d03,a  ,12345,s05
     * d03,a  ,12376,s05
     *
     */
    val curDayTemp = spark.sql(
      """
        |
        |select
        |deviceid,account,cast( count(distinct sessionid) * 100 as double) as score,max(timestamp) as timestamp
        |from cur_day_log
        |group by deviceid,account
        |
        |""".stripMargin)

    // 缓存rdd，提高重复利用该RDD的效率
    curDayTemp.cache()

    //curDayTemp.show(100, false)

    /**
     * T  cur_day
     * +--------+-------+-----+-----+
     * |deviceid|account|score|ts   |
     * +--------+-------+-----+-----+
     * |d02     |a02    |100.0|12576|
     * |d01     |null   |100.0|12376|
     * |d03     |a03    |100.0|13376|
     * |d04     |null   |100.0|14788|
     * |d01     |a01    |200.0|12368|
     * +--------+-------+-----+-----+
     *
     * T-1  pre_day
     * +--------+-------+---------+-----+
     * |deviceid|account|timestamp|score|
     * +--------+-------+---------+-----+
     * |d00     |a00    |11000    |600.0|
     * |d01     |a01    |12235    |400.0|
     * |d02     |a02    |12445    |300.0|
     * |d02     |a12    |12576    |100.0|
     * +--------+-------+---------+-----+
     *
     */

    // 整合 T-1日数据  和 T日日志得分 ==》 T日绑定结果
    // 从T日评分结果中，过滤掉account为null的数据
    curDayTemp
      .where("trim(account) != '' and account is not null")
      .createTempView("cur_day_score")

    preDayBind.createTempView("pre_day_bind")

    /**
     * 历史有，今天有，分数累加
     * 历史无，今天又，取今天
     * 历史有，今天无，取历史并衰减分数
     */
    val cur_bind_temp = spark.sql(
      """
        |
        |select
        |   nvl(pre.deviceid,cur.deviceid) as deviceid,   -- 设备id
        |   nvl(pre.account,cur.account) as account,      -- 登录账号
        |   nvl(cur.timestamp,pre.timestamp) as timestamp, -- 最近一次访问时间戳
        |   case
        |     when cur.deviceid is not null and pre.deviceid is not null then cur.score+pre.score
        |     when cur.deviceid is not null and pre.deviceid is null  then cur.score
        |     when cur.deviceid is null  and pre.deviceid is not null then pre.score*0.5
        |   end as score
        |from
        |  cur_day_score cur
        |    full join
        |  pre_day_bind pre
        |    on cur.deviceid=pre.deviceid and cur.account = pre.account
        |""".stripMargin)

    cur_bind_temp.show(100,false)


    /**
     * 将今日的新设备补充到最终结果
     */
    curDayTemp.where("account is null or trim(account)=''").createTempView("cur_maby_new")

    // 注册当日绑定临时结果表
    cur_bind_temp.createTempView("cur_temp")

    /**
     * T日最终绑定评分表计算
     * 并写入hive dws18.device_account_bind  (dt=2020-12-11)
     */
    val cur_bind_final = spark.sql(
      s"""
        |INSERT INTO TABLE dws18.device_account_bind PARTITION(dt='${args(2)}')
        |
        |select
        |  a.deviceid,
        |  a.account,
        |  a.score,
        |  a.timestamp
        |from  cur_maby_new a left join  pre_day_bind b on a.deviceid=b.deviceid
        |where b.deviceid is null
        |
        |union all
        |
        |SELECT
        |  deviceid,account,score,timestamp
        |from  cur_temp
        |
        |""".stripMargin)

    spark.close()
  }

}
